/*******************************************************************************
 * Package: com.hadoop.ip
 * Type:    IpApp
 * Date:    2024-05-17 22:25
 *
 * Copyright (c) 2024 LTD All Rights Reserved.
 *
 * You may not use this file except in compliance with the License.
 *******************************************************************************/
package com.hadoop.etl;

import com.hadoop.utils.IPParser;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.File;
import java.io.IOException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 功能描述：
 * info 字段太多  我们只取自己想要的字段 进行存储
 * 简化 必要字段  提供方便提取的输入文件
 *  ip
 *  country 国家
 *  province 省份
 *  city 城市
 *  url  url
 *  time 时间
 *  pageId id
 * @author Songxianyang
 * @date 2024-05-17 22:25
 */
public class EtlApp {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        Job job = Job.getInstance();
        // 设置启动类
        job.setJarByClass(EtlApp.class);

        // 设置 map 、reduce 类
        job.setMapperClass(EtlMapper.class);

        // map 输出 类型
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(Text.class);

        // 创建输入目录
        Path inputPath = new Path("D:\\work\\project\\gitee-github\\hadoop\\input\\act\\info.txt");
        // 创建输出目录
        Path outputPath = new Path("D:\\work\\project\\gitee-github\\hadoop\\output\\etl\\");

        // 如果存在就删除
        File folder = new File(outputPath.toString());
        if (folder.exists()) {
            deleteFiles(folder);
        }

        // 设置 FileInputFormat   FileOutputFormat
        FileInputFormat.setInputPaths(job, inputPath);
        FileOutputFormat.setOutputPath(job, outputPath);
        job.waitForCompletion(true);

    }


    private static void deleteFiles(File file) {
        if (file.isDirectory()) {
            // 如果是文件夹，则遍历其中的文件并递归删除
            for (File subFile : file.listFiles()) {
                deleteFiles(subFile);
            }
        }

        // 删除文件或空文件夹
        if (file.delete()) {
            System.out.println("已成功删除：" + file.getAbsolutePath());
        } else {
            System.out.println("删除失败：" + file.getAbsolutePath());
        }
    }

    static class EtlMapper extends Mapper<LongWritable, Text, NullWritable, Text> {

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] split = value.toString().split("\001");
            String ip = split[13];
            String url = split[1];
            String topicId = getPageId(url);
            IPParser instance= IPParser.getInstance();
            IPParser.RegionInfo info = instance.analyseIp(ip);
            String time = split[17];


            StringBuilder builder = new StringBuilder();
            builder.append(ip).append("\t");
            builder.append(info.getCountry()).append("\t");
            builder.append(info.getProvince()).append("\t");
            builder.append(info .getCity()).append("\t");
            builder.append(url).append("\t");
            builder.append(time).append("\t");
            builder.append(topicId);




            context.write(NullWritable.get(), new Text(builder.toString()));
        }
        // 获取url topicid
        private String getPageId(String url) {
            /*
            Pattern.compile("topicId=[0-9]+")：这行代码创建了一个正则表达式模式（pattern），用来匹配字符串中以"topicId="开头，后面跟着一个或多个数字的部分。
            Matcher matcher = pat.matcher(url)：这行代码使用上面创建的正则表达式模式来创建一个 Matcher 对象，并将其应用到给定的 URL 字符串上。
            if (matcher.find()) { ... }：这里判断是否在 URL 中找到了与正则表达式模式匹配的部分。
            pageId = matcher.group().split("topicId=")[1]：如果找到了匹配的部分，就提取出匹配的整个部分，然后使用 split("topicId=") 方法将其按照"topicId="进行分割，最后取得分割后的第二部分（即数字部分），并将其赋值给变量 pageId。
            */
            Pattern compile = Pattern.compile("topicId=[0-9]+");
            Matcher matcher = compile.matcher(url);
            String pageId = "-";
            if (matcher.find()) {
                pageId = matcher.group().split("topicId=")[1];
            } else {
                pageId = "-";
            }
            return pageId;
        }
    }


}
